package top.doe.spark.netty

import top.doe.spark.network.server.TransportServer
import top.doe.spark.network.{RpcHandler, TransportContext}
import top.doe.spark.rpc.{AbortableRpcFuture, RpcEndpoint, RpcEndpointRef, RpcEnv, RpcTimeout}
import top.doe.spark.util.RpcAddress

import java.util.concurrent.{Executors, ScheduledExecutorService, TimeUnit}
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.{Promise, TimeoutException}
import scala.reflect.ClassTag
import scala.util._




class NettyRpcEnv(val numUsableCores: Int) extends RpcEnv {

  var server: TransportServer = _
  lazy val address: RpcAddress = if(server != null) RpcAddress(server.host, server.port) else null
  private val timeoutScheduler: ScheduledExecutorService = Executors.newScheduledThreadPool(1)

  private val dispatcher = new Dispatcher(this, numUsableCores)
  private val transportContext = new TransportContext(new RpcHandler)
  override def setupEndpoint(name: String, endpoint: RpcEndpoint): RpcEndpointRef = {
    dispatcher.registerRpcEndpoint(name, endpoint)
  }

  def startServer(bindAddress: String, port: Int) = {
    server = transportContext.createServer(bindAddress, port)
  }

  //将消息发送出去：1.内部消息 2.远端的消息
  def send(message: RequestMessage): Unit = {
    val remoteAddr = message.receiver.address
    if(remoteAddr == address) {
      //本地的内部消息
      dispatcher.postOneWayMessage(message)
    } else {
      //TODO 远端消息

    }
  }

  //发送同步消息，消息发送成功有回调，发送失败也要有回调
  def askAbortable[T: ClassTag](message: RequestMessage, timeout: RpcTimeout): AbortableRpcFuture[T] = {
    val promise = Promise[Any]()
    //获取远端的地址
    val remoteAddr = message.receiver.address

    def onSuccess(reply: Any): Unit = {
      //将商品的promise设置为成功状态
      if(!promise.trySuccess(reply)) {
        println("ignored message: " + reply)
      }
    }

    def onFailure(e: Throwable): Unit = {
      if(!promise.tryFailure(e)) {
        println("ignored failure: " + e)
      }
    }

    if(remoteAddr == address) {
      //本地的同步消息
      val p = Promise[Any]()
      p.future.onComplete{
        case Success(response) =>  onSuccess(response)
        case Failure(e) => onFailure(e)
      }
      //通过dispatcher发送本地消息
      dispatcher.postLocalMessage(message, p)
    } else {
      //远端的同步消息
    }

    //启动定时器，进程在指定的时间内，消息是否成功被调用
    val timeoutCancelable = timeoutScheduler.schedule(
      new Runnable {
        override def run(): Unit = {
          onFailure(new TimeoutException("消息请求超时，超时的时间：" + timeout.duration))
        }
      },
      timeout.duration.toMillis, //拿出来设置好的超时时间
      TimeUnit.MILLISECONDS //时间单位
    )
    promise.future.onComplete{ v =>
      timeoutCancelable.cancel(true)
    }

    //返回AbortableRpcFuture，里面持有着Future
    new AbortableRpcFuture[T](promise.future.mapTo[T].recover(timeout.addMessageIfTimeout), onFailure)
  }


  override def endpointRef(endpoint: RpcEndpoint): RpcEndpointRef = ???

  override def setupEndpointRef(address: RpcAddress, endpointName: String): RpcEndpointRef = ???
}
